
/*
 * Step2HadoopReducer.java
 *
 * Created on May 9, 2012, 11:28:49 PM
 */

package Step2;


import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Set;
import java.util.StringTokenizer;
// import org.apache.commons.logging.Log;
// import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;

import BasicTypes.Constants;
import UnionFind.UnionFindHashTable;

/**
 *
 * @author crazymooner
 */
public class Step2HadoopReducer extends MapReduceBase implements Reducer<IntWritable,Text,IntWritable,IntWritable> {
    // The Karmasphere Studio Workflow Log displays logging from Apache Commons Logging, for example:
    // private static final Log LOG = LogFactory.getLog("Step2.Step2HadoopReducer");

	@Override
    public void reduce(IntWritable key, Iterator<Text> value, OutputCollector<IntWritable, IntWritable> output, Reporter reporter)
            throws IOException {
		   UnionFindHashTable uf = new UnionFindHashTable();
	       //QuickUnion qu = new QuickUnion((Constants.COLUMESIZE + 1) * Constants.SIZE, group);
	       while (value.hasNext()) {
	    	StringTokenizer st = new StringTokenizer(value.next().toString());
			int p = Integer.parseInt(st.nextToken());
			int q = Integer.parseInt(st.nextToken());
			uf.unite(p, q);
	       }
	       
	       Enumeration<Integer> keys = uf.getMetrix().keys();
	       
	       while( keys.hasMoreElements() )
	       {
	    	   int aKey = keys.nextElement();
	    	   int aValue = uf.getMetrix().get(aKey);
	    	   output.collect(new IntWritable(aKey), new IntWritable(aValue));
	       }
    }
}
